package cn.doitedu.dashboard;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job04_PerBrandTopnProduct {
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        TableConfig config = tenv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled", true); // 开启窗口提前触发器
        config.getConfiguration().setString("table.exec.emit.early-fire.delay", "1 s");  // 每隔1s周期性触发


        // 建表，映射  oms_order
        tenv.executeSql(
                " create table  oms_order_mysql(             "+
                        " id bigint,                         "+
                        " status int,                        "+  // 订单状态; 0未支付，1待发货 ，2已发货，3已完成
                        " create_time timestamp(3),          "+   // 订单创建时间
                        " payment_time timestamp(3),          "+   // 订单支付时间
                        " modify_time timestamp(3),          "+   // 数据更新时间
                        " rt as  modify_time,                "+
                        " watermark for rt as rt,            "+
                        " primary key (id) not enforced      "+
                        " ) with (                           "+
                        "     'connector' = 'mysql-cdc',     "+
                        "     'hostname' = 'doitedu',        "+
                        "     'port' = '3306',               "+
                        "     'username' = 'root',           "+
                        "     'password' = 'root',           "+
                        "     'database-name' = 'realtimedw',  "+
                        "     'table-name' = 'oms_order'      "+
                        " )                                    "
        );

        // 建表，映射  oms_order_item
        tenv.executeSql(
                " create table  oms_order_item_mysql(        "+
                        " id   bigint,                       "+  // 自增主键，无意义
                        " order_id bigint,                   "+  // 所属的订单id
                        " product_id bigint,                 "+  // 商品id
                        " product_name string,               "+  // 商品名称
                        " product_brand string,              "+  // 商品所属的品牌
                        " real_amount decimal(10,2),         "+  // 商品在本订单中的实付额
                        " primary key (id) not enforced      "+
                        " ) with (                           "+
                        "     'connector' = 'mysql-cdc',     "+
                        "     'hostname' = 'doitedu',        "+
                        "     'port' = '3306',               "+
                        "     'username' = 'root',           "+
                        "     'password' = 'root',           "+
                        "     'database-name' = 'realtimedw',  "+
                        "     'table-name' = 'oms_order_item'      "+
                        " )                                    "
        );


        // 关联
        tenv.executeSql(
                       " CREATE TEMPORARY VIEW joined_view AS                                                "+
                        " SELECT                                                                              "+
                        "     a.order_id,                                                                     "+
                        "     a.order_status,                                                                 "+
                        "     a.rt,                                                                           "+
                        "     b.product_id,                                                                   "+
                        "     b.product_name,                                                                 "+
                        "     b.product_brand,                                                                "+
                        "     b.real_amount                                                                   "+
                        " from                                                                                "+
                        " (                                                                                   "+
                        "     select                                                                          "+
                        "         id as order_id,                                                             "+
                        "         status as order_status,                                                     "+
                        "         rt                                                                          "+
                        "     from oms_order_mysql                                                            "+
                        "     where status=1 and date_format(payment_time,'yyyy-MM-dd')=current_date    "+
                        " ) a                                                                                 "+
                        " join oms_order_item_mysql b                                                         "+
                        " on a.order_id = b.order_id                                                          "
        );


        // 求分组topn
        tenv.executeSql(
                        " WITH tmp1 AS (                                                                                                             "+
                        " SELECT                                                                                                                     "+
                        "     TUMBLE_START(rt,INTERVAL '24' HOUR) as window_start,                                                                   "+
                        "     TUMBLE_END(rt,INTERVAL '24' HOUR) as window_end,                                                                       "+
                        "     product_brand,                                                                                                         "+
                        "     product_id,                                                                                                            "+
                        "     product_name,                                                                                                          "+
                        "     SUM(real_amount) AS real_amount                                                                                         "+
                        " FROM  joined_view                                                                                                          "+
                        " GROUP BY                                                                                                                   "+
                        "     TUMBLE(rt,INTERVAL '24' HOUR),                                                                                         "+
                        "     product_brand,                                                                                                         "+
                        "     product_id,                                                                                                            "+
                        "     product_name  )                                                                                                        "+
                        "                                                                                                                            "+
                        " ,tmp2 AS (                                                                                                                 "+
                        " SELECT                                                                                                                     "+
                        "     window_start,                                                                                                          "+
                        "     window_end,                                                                                                            "+
                        "     product_brand,                                                                                                         "+
                        "     product_id,                                                                                                            "+
                        "     product_name,                                                                                                          "+
                        "     real_amount,                                                                                                           "+
                        "     row_number() over(PARTITION BY window_start,window_end,product_brand ORDER BY real_amount DESC) as rn                  "+
                        " FROM tmp1 )                                                                                                                "+
                        "                                                                                                                            "+
                        "                                                                                                                            "+
                        " SELECT                                                                                                                     "+
                        "     window_start,                                                                                                          "+
                        "     window_end,                                                                                                            "+
                        "     product_brand,                                                                                                         "+
                        "     product_id,                                                                                                            "+
                        "     product_name,                                                                                                          "+
                        "     real_amount,                                                                                                           "+
                        "     rn                                                                                                                     "+
                        " FROM tmp2                                                                                                                  "+
                        " WHERE rn<=2                                                                                                                "
        ).print();


    }
}
